package com.tcm.kafka;

import io.protostuff.LinkedBuffer;
import io.protostuff.ProtostuffIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;
import org.apache.kafka.common.serialization.Serializer;

public class CompanyProtostuffSerializer implements Serializer<Company> {

    @Override
    public byte[] serialize(String topic, Company company) {
        if(company == null) return null;

        Schema schema = RuntimeSchema.getSchema(company.getClass());
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        byte[] protostuff = null;
        try{
            protostuff = ProtostuffIOUtil.toByteArray(company, schema, buffer);
        }catch (Exception e){
            e.printStackTrace();
        }finally{
            buffer.clear();
        }
        return protostuff;
    }
}
